/*
 * Copyright (C) 2013 Square, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package okhttp3;

import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Deque;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import javax.annotation.Nullable;

import okhttp3.RealCall.AsyncCall;
import okhttp3.internal.Util;

/**
 * Policy on when async requests are executed.
 *
 * <p>
 * Each dispatcher uses an {@link ExecutorService} to run calls internally. If
 * you supply your own executor, it should be able to run
 * {@linkplain #getMaxRequests the configured maximum} number of calls
 * concurrently.
 */
public final class Dispatcher {
	private int maxRequests = 64; // 最大并发请求数
	private int maxRequestsPerHost = 5; // 每个主机的最大请求数
	private @Nullable Runnable idleCallback;

	/** Executes calls. Created lazily. */
	// 消费者线程池
	private @Nullable ExecutorService executorService;

	/** Ready async calls in the order they'll be run. */
	// Deque接口具有丰富的抽象数据形式，它支持从队列两端点检索和插入元素
	// 为何要使用这两个队列呢？把Dispatcher当成生产者，把线程池当成消费者，当生产者生产的线程大于消费者所能承受的最大范围，就把未能及时执行的任务保存在readyAsyncCalls队列中，当时机成熟，也就是线程池有空余线程可以执行时，会调用promoteCall()这个方法把等待队列中的任务取出放到线程池中执行，并且把这个任务转移到runningAsyncCalls队列中去
	// 将要运行的异步请求队列
	private final Deque<AsyncCall> readyAsyncCalls = new ArrayDeque<>();
	/**
	 * Running asynchronous calls. Includes canceled calls that haven't finished
	 * yet.
	 */
	// 正在请求的异步请求队列
	private final Deque<AsyncCall> runningAsyncCalls = new ArrayDeque<>();

	/**
	 * Running synchronous calls. Includes canceled calls that haven't finished yet.
	 */
	// 正在请求的同步请求队列
	private final Deque<RealCall> runningSyncCalls = new ArrayDeque<>();

	// Dispatcher有两个构造方法，可以使用自己设定的线程池。如果没有设定线程池，则会在请求网络前创建默认的线程池
	public Dispatcher(ExecutorService executorService) {
		this.executorService = executorService;
	}

	public Dispatcher() {
	}

	public synchronized ExecutorService executorService() {
		if (executorService == null) {
			// 可以看到调度器Dispatcher内部维护了一个ThreadPoolExecutor线程池，并直接将call对象传入线程池执行
			executorService = new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60, TimeUnit.SECONDS,
					new SynchronousQueue<Runnable>(), Util.threadFactory("OkHttp Dispatcher", false));
		}
		return executorService;
	}

	/**
	 * Set the maximum number of requests to execute concurrently. Above this
	 * requests queue in memory, waiting for the running calls to complete.
	 *
	 * <p>
	 * If more than {@code maxRequests} requests are in flight when this is invoked,
	 * those requests will remain in flight.
	 */
	public void setMaxRequests(int maxRequests) {
		if (maxRequests < 1) {
			throw new IllegalArgumentException("max < 1: " + maxRequests);
		}
		synchronized (this) {
			this.maxRequests = maxRequests;
		}
		promoteAndExecute();
	}

	public synchronized int getMaxRequests() {
		return maxRequests;
	}

	/**
	 * Set the maximum number of requests for each host to execute concurrently.
	 * This limits requests by the URL's host name. Note that concurrent requests to
	 * a single IP address may still exceed this limit: multiple hostnames may share
	 * an IP address or be routed through the same HTTP proxy.
	 *
	 * <p>
	 * If more than {@code maxRequestsPerHost} requests are in flight when this is
	 * invoked, those requests will remain in flight.
	 *
	 * <p>
	 * WebSocket connections to hosts <b>do not</b> count against this limit.
	 */
	public void setMaxRequestsPerHost(int maxRequestsPerHost) {
		if (maxRequestsPerHost < 1) {
			throw new IllegalArgumentException("max < 1: " + maxRequestsPerHost);
		}
		synchronized (this) {
			this.maxRequestsPerHost = maxRequestsPerHost;
		}
		promoteAndExecute();
	}

	public synchronized int getMaxRequestsPerHost() {
		return maxRequestsPerHost;
	}

	/**
	 * Set a callback to be invoked each time the dispatcher becomes idle (when the
	 * number of running calls returns to zero).
	 *
	 * <p>
	 * Note: The time at which a {@linkplain Call call} is considered idle is
	 * different depending on whether it was run {@linkplain Call#enqueue(Callback)
	 * asynchronously} or {@linkplain Call#execute() synchronously}. Asynchronous
	 * calls become idle after the {@link Callback#onResponse onResponse} or
	 * {@link Callback#onFailure onFailure} callback has returned. Synchronous calls
	 * become idle once {@link Call#execute() execute()} returns. This means that if
	 * you are doing synchronous calls the network layer will not truly be idle
	 * until every returned {@link Response} has been closed.
	 */
	public synchronized void setIdleCallback(@Nullable Runnable idleCallback) {
		this.idleCallback = idleCallback;
	}

	void enqueue(AsyncCall call) {
		synchronized (this) {
			// 将当前的Call添加到异步队列中，并会在请求执行完毕的时候从该队列中移除，只是这里的队列是runningAsyncCalls或者readyAsyncCalls。它们都是一个双端队列，并用来存储异步类型的请求
			// 它们的区别是，runningAsyncCalls 是正在执行的队列，当正在执行的队列达到了限制的时候，就会将其放置到就绪队列
			// readyAsyncCalls 中
			readyAsyncCalls.add(call);
		}
		promoteAndExecute();
	}

	/**
	 * Cancel all calls currently enqueued or executing. Includes calls executed
	 * both {@linkplain Call#execute() synchronously} and {@linkplain Call#enqueue
	 * asynchronously}.
	 */
	public synchronized void cancelAll() {
		for (AsyncCall call : readyAsyncCalls) {
			call.get().cancel();
		}

		for (AsyncCall call : runningAsyncCalls) {
			call.get().cancel();
		}

		for (RealCall call : runningSyncCalls) {
			call.cancel();
		}
	}

	/**将符合条件的调用从readyAsyncCalls提升到runningAsyncCalls，并在executor服务上运行它们。不能以同步方式调用，因为执行调用会调用用户代码
	 * <br/>Promotes eligible calls from {@link #readyAsyncCalls} to
	 * {@link #runningAsyncCalls} and runs them on the executor service. Must not be
	 * called with synchronization because executing calls can call into user code.
	 *
	 * @return true if the dispatcher is currently running calls.
	 */
	private boolean promoteAndExecute() {
		assert (!Thread.holdsLock(this));

		List<AsyncCall> executableCalls = new ArrayList<>();
		boolean isRunning;
		synchronized (this) {
			for (Iterator<AsyncCall> i = readyAsyncCalls.iterator(); i.hasNext();) {
				AsyncCall asyncCall = i.next();

				// 如果正在运行的异步请求不超过64，而且同一个host下的异步请求不得超过5个则将请求添加到正在运行的同步请求队列中runningAsyncCalls并开始
				// 执行请求，否则就添加到readyAsyncCalls继续等待
				if (runningAsyncCalls.size() >= maxRequests)
					break; // Max capacity.
				if (runningCallsForHost(asyncCall) >= maxRequestsPerHost)
					continue; // Host max capacity.

				i.remove();
				executableCalls.add(asyncCall);
				runningAsyncCalls.add(asyncCall);
			}
			isRunning = runningCallsCount() > 0;
		}

		for (int i = 0, size = executableCalls.size(); i < size; i++) {
			AsyncCall asyncCall = executableCalls.get(i);
			asyncCall.executeOn(executorService());
		}

		return isRunning;
	}

	/** Returns the number of running calls that share a host with {@code call}. */
	private int runningCallsForHost(AsyncCall call) {
		int result = 0;
		for (AsyncCall c : runningAsyncCalls) {
			if (c.get().forWebSocket)
				continue;
			if (c.host().equals(call.host()))
				result++;
		}
		return result;
	}

	/** Used by {@code Call#execute} to signal it is in-flight. */
	synchronized void executed(RealCall call) {
		// RealCall将其加入到双端队列中，请注意这里的双端队列的名称是 runningSyncCalls，也就是说这种请求是同步请求，会在当前的线程中立即被执行
		// 下面的 getResponseWithInterceptorChain() 就是这个同步的执行过程。而当我们执行完毕的时候，又会调用 Dispatcher 的 finished(RealCall) 方法把该请求从队列中移除
		runningSyncCalls.add(call);
	}

	/** Used by {@code AsyncCall#run} to signal completion. */
	void finished(AsyncCall call) {
		finished(runningAsyncCalls, call);
	}

	/** Used by {@code Call#execute} to signal completion. */
	void finished(RealCall call) {
		finished(runningSyncCalls, call);
	}

	private <T> void finished(Deque<T> calls, T call) {
		Runnable idleCallback;
		synchronized (this) {
			if (!calls.remove(call))
				throw new AssertionError("Call wasn't in-flight!");
			idleCallback = this.idleCallback;
		}

		boolean isRunning = promoteAndExecute();

		if (!isRunning && idleCallback != null) {
			idleCallback.run();
		}
	}

	/** Returns a snapshot of the calls currently awaiting execution. */
	public synchronized List<Call> queuedCalls() {
		List<Call> result = new ArrayList<>();
		for (AsyncCall asyncCall : readyAsyncCalls) {
			result.add(asyncCall.get());
		}
		return Collections.unmodifiableList(result);
	}

	/** Returns a snapshot of the calls currently being executed. */
	public synchronized List<Call> runningCalls() {
		List<Call> result = new ArrayList<>();
		result.addAll(runningSyncCalls);
		for (AsyncCall asyncCall : runningAsyncCalls) {
			result.add(asyncCall.get());
		}
		return Collections.unmodifiableList(result);
	}

	public synchronized int queuedCallsCount() {
		return readyAsyncCalls.size();
	}

	public synchronized int runningCallsCount() {
		return runningAsyncCalls.size() + runningSyncCalls.size();
	}
}
